package com.lenovo.spider;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ArrayUtil;
import com.lenovo.spider.config.ConfigHolder;
import com.lenovo.spider.domain.Request;
import com.lenovo.spider.domain.Task;
import com.lenovo.spider.process.downloader.Downloader;
import com.lenovo.spider.process.downloader.HttpDownloader;
import com.lenovo.spider.process.exceptions.ChainExceptionHandler;
import com.lenovo.spider.process.exceptions.DefaultExceptionHandler;
import com.lenovo.spider.process.parser.PageParser;
import com.lenovo.spider.process.pipeline.ChainPipeline;
import com.lenovo.spider.process.pipeline.ConsolePipeline;
import com.lenovo.spider.process.pipeline.Pipeline;
import com.lenovo.spider.queue.TaskQueue;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import reactor.core.publisher.WorkQueueProcessor;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

import static com.lenovo.spider.support.Functions.monoWrap;
import static com.lenovo.spider.support.Functions.when;

/**
 * @author <a href="mailto:liuhan3@lenovo.com">HanL(liuhan3)</a>
 * on 2018/3/20 11:16
 */
@Slf4j
public class Spider {
    public static final int WAIT = 0;
    public static final int RUNNING = 1;
    public static final int STOP = 2;
    /**
     * 工作流处理器
     */
    private final WorkQueueProcessor<Task> processor;
    /**
     * 任务队列
     */
    private TaskQueue taskQueue;
    /**
     * 下载器
     */
    private Downloader downloader;
    /**
     * 页面解析器
     */
    private PageParser pageParser;
    /**
     * 结果处理管道
     */
    private ChainPipeline pipeline;
    /**
     * 异常处理器
     */
    private ChainExceptionHandler exceptionHandler;
    /**
     * 爬虫状态
     */
    private AtomicInteger state;


    public Spider() {
        val spiderName = ConfigHolder.getString("spider.name", Spider.class.getSimpleName());
        val bufferSize = ConfigHolder.getInt("spider.bufferSize", 1024);
        this.processor = WorkQueueProcessor.create(spiderName, bufferSize);
        configProcessor();
        this.state = new AtomicInteger(0);
    }

    /**
     * 配置爬虫处理工作流
     */
    public void configProcessor() {
        this.processor.delayElements(Duration.ofMillis(500))
                .log()
                .flatMap(monoWrap(this.downloader::download))  // 下载
                .flatMap(monoWrap(this.pageParser::parser)) // 页面解析
                .subscribe(when(Task::isSuccess,
                        this.pipeline::pipeline, // 结果处理
                        this.exceptionHandler::handle  //异常处理
                ));
    }


    /**
     * 获取爬虫执行体
     *
     * @return
     */
    public Runnable getRunnable() {
        return () -> {
            this.pushStartRequests();
            this.setRunning(); // 设置启动状态
            val key = this.pageParser.getSiteConfig().getIdentity();
            // 循环读取任务队列，并压入流处理器
            while (this.isRunning()) {
                try {

                    val request = this.taskQueue.poll(key);
                    if (request == null) {
                        ThreadUtil.safeSleep(1000);
                    }
                    this.processor.onNext(Task.create(request));
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }

            // 爬虫停止后，等待流处理器中的处理完成后再关闭流处理器
            this.processor.awaitAndShutdown();
        };
    }

    /**
     * 停止
     */
    public void stop() {
        for (; ; ) {
            if (this.state.compareAndSet(RUNNING, STOP)) {
                break;
            }
        }
    }

    /**
     * 判断当前状态是否正在运行
     *
     * @return
     */
    public boolean isRunning() {
        return this.state.get() == RUNNING;
    }

    /**
     * 判断当前状态是否为停止
     *
     * @return
     */
    public boolean isStop() {
        return this.state.get() == STOP;
    }

    /**
     * 起始请求压入任务队列中
     */
    private void pushStartRequests() {
        val siteConfig = this.pageParser.getSiteConfig();
        val requests = siteConfig.getStartRequests();
        if (CollUtil.isNotEmpty(requests)) {
            val key = siteConfig.getIdentity();
            this.taskQueue.push(key, ArrayUtil.toArray(requests, Request.class));
        }
    }

    /**
     * 设置运行状态
     */
    private void setRunning() {
        for (; ; ) {
            if (this.state.compareAndSet(WAIT, RUNNING)) {
                break;
            }
        }
    }

    public static void main(String[] args) throws IOException {
        WorkQueueProcessor<Task> processor = WorkQueueProcessor.create("spider", 1024);


        for (int i = 0; i < 3; i++) {
            val request = Request.builder().url("http://www.baidu.com/" + i)
                    .downloaderType(i % 2 == 0 ? Request.DownloaderType.HTTP : Request.DownloaderType.WEBDRIVER)
                    .build();
            processor.onNext(Task.create(request));

        }
        System.out.println(new BufferedReader(new InputStreamReader(System.in)).readLine());
    }
}
